package com.googlecode.jbeansdb.impl;

import java.io.IOException;
import java.nio.channels.SocketChannel;
import java.util.Arrays;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;

import com.googlecode.jbeansdb.Beansdb;
import com.googlecode.jbeansdb.RequestProcessor;
import com.googlecode.jbeansdb.StorageEngine;

public class DefaultRequestProcessor implements RequestProcessor {
	private static final Logger log = Logger.getLogger(DefaultRequestProcessor.class);
	private StorageEngine engine;
	private String op;
	private String key;
	private int valueLen;
	private MemcachedDecoder memDecorder = MemcachedDecoder.getInstance();
	private BlockingQueue<ServerDataEvent> queue = new LinkedBlockingQueue<ServerDataEvent>();
	
	public DefaultRequestProcessor(Beansdb db) {
		this.engine = db.getEngine();
	}
	
	public void process(Beansdb beansdb, SocketChannel socket, byte[] datainput, int count) throws IOException, InterruptedException {
		byte[] data = new byte[count];
		System.arraycopy(datainput, 0, data, 0, count);
		log.info("process:data is: " + new String(data));
		String command = memDecorder.getCommandName(data);
		
		log.info("command: " + command + ", op:" + this.op);
		if ("get".equals(command)) {
			String key = memDecorder.getGetKey(command, data);
			
			this.op = "get";
			this.key = key;
			
			log.info("GET " + key);
			String value = (String)this.engine.get(key);
			log.info("GOT {" + key + ":" + value + "}");
			if (value == null) {
				value = "just for test.";
			}
			this.queue.put(new ServerDataEvent(beansdb, socket, value.getBytes()));
		} else if ("set".equals(command)) {
			String key = memDecorder.getSetKey(command, data);
			this.op = "set";
			this.key = key;
			this.valueLen = memDecorder.getValueLength(data);
			log.info("SET1 " + this.key + "'s value length: " + this.valueLen);
		} else if ("delete".equals(command)) {
			String key = memDecorder.getGetKey(command, data);			
			this.op = "delete";
			this.key = key;
			
			log.info("DELETE " + key + "'s value");
			this.engine.delete(key);
		} else if ("set".equals(this.op)){
			String value = new String(Arrays.copyOfRange(data, 0, this.valueLen));
			log.info("SET2 " + this.key + "'s value: " + value);
			this.engine.put(this.key, value);

			this.queue.put(new ServerDataEvent(beansdb, socket, "STORED\r\n".getBytes()));
		} else {
			// empty
			log.error("a message got into else... Orz");
		}
	}

	public void run() {
		ServerDataEvent dataEvent = null;
		while(true) {
			try {
				dataEvent = (ServerDataEvent) queue.take();
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			
			// Return to sender
			dataEvent.server.send(dataEvent.socket, dataEvent.data);
		}
	}
	
	class ServerDataEvent {
		public Beansdb server;
		public SocketChannel socket;
		public byte[] data;
		
		public ServerDataEvent(Beansdb server, SocketChannel socket, byte[] data) {
			this.server = server;
			this.socket = socket;
			this.data = data;
		}
	}
}
